package com.jython.flink;

import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.SplitStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

import java.util.ArrayList;

public class WordCount2 {
    public static void main(String[] args) {
        int port;
        try {
            ParameterTool parameterTool = ParameterTool.fromArgs(args);
            port = parameterTool.getInt("port");
        }catch (Exception e){
            System.out.println("没有指定port,默认为9000");
            port=9000;
        }
        StreamExecutionEnvironment env= StreamExecutionEnvironment.getExecutionEnvironment();
//        DataStream<String> textStream= env.socketTextStream("192.168.0.180", port, "\n");
        DataStream<String> shuzi = env.fromElements("1", "2", "3", "4", "5");
        DataStream<String> zimu= env.fromElements("A", "B", "C", "D", "E");
        DataStream<Object> map = shuzi.connect(zimu).map(new CoMapFunction<String, String, Object>() {
            @Override
            public Object map1(String s) throws Exception {
//                System.out.println("--" + s);
                return s;
            }

            @Override
            public Object map2(String s) throws Exception {
//                System.out.println("++" + s);
                return s;
            }
        });
        map.map(x->{
            System.out.println(x);
            return x;
        });
        try {
            env.execute("ok");
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
